package com.gd.data.platform.flink.jobs;

import java.util.Map;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.gd.data.platform.flink.utils.FlinkUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * 实时指标计算作业
 * 使用Flink进行实时数据计算
 * 
 * @author gd-data-platform
 * @since 1.0.0
 */
@Slf4j
@Component
public class RealtimeMetricsJob {

    @Autowired
    private StreamExecutionEnvironment streamExecutionEnvironment;

    @Autowired
    private KafkaSource<String> kafkaConsumer;

    @Autowired
    private KafkaSink<String> kafkaProducer;

    /**
     * 启动实时计算作业
     */
    public void startJob() throws Exception {
        log.info("启动实时指标计算作业");

        // 从Kafka读取API调用日志
        DataStream<String> apiCallLogs = streamExecutionEnvironment
                .fromSource(kafkaConsumer, org.apache.flink.api.common.eventtime.WatermarkStrategy.noWatermarks(),
                        "API调用日志源");

        // 解析API调用日志
        DataStream<Map<String, Object>> parsedLogs = apiCallLogs
                .map(new ParseApiCallLogFunction())
                .name("解析API调用日志");

        // 计算调用量指标
        SingleOutputStreamOperator<String> callCountMetrics = parsedLogs
                .keyBy(log -> (String) log.get("apiName"))
                .window(SlidingProcessingTimeWindows.of(Time.seconds(60), Time.seconds(10)))
                .process(new CallCountProcessFunction())
                .name("计算调用量指标");

        // 计算失败率指标
        SingleOutputStreamOperator<String> failureRateMetrics = parsedLogs
                .keyBy(log -> (String) log.get("apiName"))
                .window(SlidingProcessingTimeWindows.of(Time.seconds(60), Time.seconds(10)))
                .process(new FailureRateProcessFunction())
                .name("计算失败率指标");

        // 计算IP访问指标
        SingleOutputStreamOperator<String> ipAccessMetrics = parsedLogs
                .keyBy(log -> (String) log.get("clientIp"))
                .window(SlidingProcessingTimeWindows.of(Time.seconds(60), Time.seconds(10)))
                .process(new IpAccessProcessFunction())
                .name("计算IP访问指标");

        // 合并所有指标
        DataStream<String> allMetrics = callCountMetrics
                .union(failureRateMetrics)
                .union(ipAccessMetrics);

        // 发送到Kafka
        allMetrics.sinkTo(kafkaProducer).name("发送指标到Kafka");

        // 启动作业
        streamExecutionEnvironment.execute("RealtimeMetricsJob");
    }

    /**
     * 解析API调用日志函数
     */
    public static class ParseApiCallLogFunction implements MapFunction<String, Map<String, Object>> {
        @Override
        public Map<String, Object> map(String json) throws Exception {
            return FlinkUtils.parseJson(json);
        }
    }

    /**
     * 调用量计算函数
     */
    public static class CallCountProcessFunction
            extends ProcessWindowFunction<Map<String, Object>, String, String, TimeWindow> {
        @Override
        public void process(String key, Context context, Iterable<Map<String, Object>> elements, Collector<String> out)
                throws Exception {
            long callCount = 0;
            for (Map<String, Object> element : elements) {
                // 统计调用次数
                callCount++;
            }

            Map<String, Object> metric = FlinkUtils.createRealtimeMetric(
                    "call_count", "API调用量", (double) callCount, "{\"apiName\":\"" + key + "\"}");

            out.collect(FlinkUtils.toJson(metric));
            log.info("API {} 调用量：{}", key, callCount);
        }
    }

    /**
     * 失败率计算函数
     */
    public static class FailureRateProcessFunction
            extends ProcessWindowFunction<Map<String, Object>, String, String, TimeWindow> {
        @Override
        public void process(String key, Context context, Iterable<Map<String, Object>> elements, Collector<String> out)
                throws Exception {
            long totalCount = 0;
            long failureCount = 0;

            for (Map<String, Object> log : elements) {
                totalCount++;
                Boolean success = (Boolean) log.get("success");
                if (success != null && !success) {
                    failureCount++;
                }
            }

            double failureRate = totalCount > 0 ? (double) failureCount / totalCount : 0.0;

            Map<String, Object> metric = FlinkUtils.createRealtimeMetric(
                    "failure_rate", "API失败率", failureRate, "{\"apiName\":\"" + key + "\"}");

            out.collect(FlinkUtils.toJson(metric));
            log.info("API {} 失败率：{}", key, failureRate);
        }
    }

    /**
     * IP访问计算函数
     */
    public static class IpAccessProcessFunction
            extends ProcessWindowFunction<Map<String, Object>, String, String, TimeWindow> {
        @Override
        public void process(String key, Context context, Iterable<Map<String, Object>> elements, Collector<String> out)
                throws Exception {
            long accessCount = 0;
            for (Map<String, Object> element : elements) {
                // 统计访问次数
                accessCount++;
            }

            Map<String, Object> metric = FlinkUtils.createRealtimeMetric(
                    "ip_access", "IP访问量", (double) accessCount, "{\"clientIp\":\"" + key + "\"}");

            out.collect(FlinkUtils.toJson(metric));
            log.info("IP {} 访问量：{}", key, accessCount);
        }
    }
}
